package cn.com.zetatech.loader.resolver.kafka.consumer;

import cn.com.zetatech.loader.entity.ParquetMessage;
import cn.com.zetatech.loader.resolver.service.AnnotationParquetFunction;
import cn.com.zetatech.loader.resolver.service.AnnotationResolverParquetFunctionFactory;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author jian.chen@zetatech.com.cn
 * @Description TODO
 * @Date 2024/8/18 14:59
 */
@Slf4j
@Component
public class AnnotationParquetResolveConsumerFac {

    @Autowired
    private Gson gson;

    @Autowired
    private AnnotationResolverParquetFunctionFactory functionFactory;

    @KafkaListener(topics = "#{'${resolver.topic}'.split(',')}",
            concurrency = "${resolver.consumer.concurrency:2}",
            groupId = "${resolver.consumer.groupId:parquet_consume3 }")
    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            String value = record.value();
            String topic = record.topic();
            String key = record.key();

            System.out.println(topic);
            System.out.println(key);
            if(!StringUtils.isEmpty(value)) {
                // 对消息进行解码
                log.info("<============================ 消费Kafka parquet消息  ============================>");
                ParquetMessage message = gson.fromJson(value, ParquetMessage.class);
                if(message != null) {
                    log.info("<============================ 开始解析任务 ============================>");

                    // 根据type获取对应的策略并执行
                    AnnotationParquetFunction function = functionFactory.getFunction(message.getType());
                    if (function != null) {
                        function.process(message);
                    } else {
                        log.warn("未找到对应的处理策略，type: " + message.getType());
                    }
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            ack.acknowledge();
        }
    }
}
